package com.gitee.window

import java.util.UUID

import com.gitee.watermaker.OrderWaterMarkDemo.Order
import org.apache.commons.lang3.time.FastDateFormat
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment,_}
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows
import org.apache.flink.streaming.api.windowing.time.Time

import scala.util.Random

//事件时间窗口
object SessionWindow {
  //订单样例类Order(订单ID、用户ID、订单金额、时间戳(事件时间))
  case class Order(orderId: String, userId: Int, money: Long, timestamp: Long, systems: Long)

  //定义一个时间转换工具
  val df: FastDateFormat = FastDateFormat.getInstance("HH:mm:ss")
  def main(args: Array[String]): Unit = {
    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //准备乱序的订单数据
    import org.apache.flink.api.scala._
    val orderDS: DataStream[Order] = senv.addSource(new RichSourceFunction[Order] {
      var isRunning = true

      override def run(ctx: SourceFunction.SourceContext[Order]): Unit = {
        while (isRunning) {
          val orderId: String = UUID.randomUUID().toString.replaceAll("-", "")
          val userId: Int = Random.nextInt(1)
          val money: Int = Random.nextInt(101)
          val timestamp: Long = System.currentTimeMillis() - Random.nextInt(20) * 1000 //减去一个随机的x秒时间表示该订单是该x秒之前产生的
          val systems: Long = System.currentTimeMillis() //系统时间
          val order = Order(orderId, userId, money, timestamp, systems)
          ctx.collect(order)
          Thread.sleep(1000)
        }
      }

      override def cancel(): Unit = {
        isRunning = false
      }
    })

    val watermarkData: DataStream[Order] = orderDS.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[Order](Time.milliseconds(3000)) {
      override def extractTimestamp(element: Order): Long = {
        element.timestamp
      }
    })

    val result: DataStream[Order] = watermarkData.keyBy(_.userId).window(EventTimeSessionWindows.withGap(Time.seconds(5))).sum(2)
    result.print()
    senv.execute()
  }

}
